package com.util;

import java.io.File;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import com.utils.queue.FilePersistentQueue;

public class PersistentQueueManager extends Singleton {
	
	public static PersistentQueueManager getInstance()
	{
		return (PersistentQueueManager) PersistentQueueManager.getInstance(PersistentQueueManager.class.getName());
	}


	private static Logger logger = Logger.getLogger(PersistentQueueManager.class);
		
	private Map<String, FilePersistentQueue> queue;
	private Map<String, Long> count;
	private Map<String, Long> wait;
		
	protected PersistentQueueManager()
	{
		queue = new HashMap<String, FilePersistentQueue>();
		count = new HashMap<String, Long>();
		wait = new HashMap<String, Long>();
	}
	
	public synchronized FilePersistentQueue getQueue(String name)
	{
		if (!queue.containsKey(name))
		{
			File dir = new File(ConfigUtil.get("manage", com.manage.config.Config.KEY_QUEUEFILEDIR)+"/" + name);
			if (!dir.exists())
				dir.mkdirs();
			logger.info("Queue Path:" + dir.getAbsolutePath());
			
			queue.put(name, new FilePersistentQueue(dir));
			count.put(name, new Long(0));
			wait.put(name, new Long(0));
		}
		return queue.get(name);
	}
	
	public boolean offer(String name, Serializable o)
	{
		FilePersistentQueue pq = getQueue(name);
		synchronized (pq) {
			if (pq.push(o))
			{
				count.put(name, count.get(name) + 1);
				if (wait.get(name) > 0)
					pq.notify();
				return true;
			}
		}
		return false;
	}
	
	public Serializable poll(String name)
	{
		FilePersistentQueue pq = getQueue(name);
		synchronized (pq) {
			try
			{
				if (pq.isEmpty())
				{
					wait.put(name, wait.get(name) + 1);
					pq.wait();
					wait.put(name, wait.get(name) - 1);
				}
			}
			catch (Exception e)
			{
				
			}
			return pq.pop();
		}
	}
	
	public String status()
	{
		String status = "";
		for (Entry<String, FilePersistentQueue> q : queue.entrySet())
		{
			String name = q.getKey();
			FilePersistentQueue p = q.getValue();
			status += "queue status [name:" + name + ";count:" + count.get(name) + ";queue:"  + p.size() + ";wait:" + wait.get(name) + "]\r\n";
		}
		return status;
	}
}
